package com.neusoft.bd.myflink.mytable.step1;


import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;

public class TableApiDemo {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        GeneratorFunction<Long, String> generatorFunction = index -> "Number: " + index;
//        long numberOfRecords = 1000;
//
//        DataGeneratorSource<String> source =
//                new DataGeneratorSource<>(generatorFunction, numberOfRecords, Types.STRING);
//        DataStreamSource<String> stream =
//                env.fromSource(source,
//                        WatermarkStrategy.noWatermarks(),
//                        "Generator Source");
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();

        TableEnvironment tableEnvironment = TableEnvironment.create(settings);
        tableEnvironment.getConfig().getConfiguration().setString("parallelism.default", "2");
        tableEnvironment.createTemporaryTable("SourceTable ", TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("f0", DataTypes.STRING())
                        .build())
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 10L)
                .build());

        tableEnvironment.executeSql("create table SinkTable " +
                "( f0 string )" +
                "with( 'connector' ='print'  )");

//        Table table1 = tableEnvironment.from("SourceTable");
//        table1.execute().print();
//        得到table对象
//        Table table2 = tableEnvironment.sqlQuery("select * from  SourceTable");
//        table2.execute().print();
//        tableEnvironment.executeSql(" select * from  SourceTable  ").print();
//        Table table1 = tableEnvironment.from("SourceTable");
//        table1.insertInto("SinkTable").execute();

        Table table1 = tableEnvironment.from("SourceTable");
        table1.executeInsert("SinkTable");

    }
}
